package com.twoandone.tms.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.twoandone.tms.realtime.beans.TableProcess;
import com.twoandone.tms.realtime.common.TmsConfig;
import com.twoandone.tms.realtime.utils.DruidDSUtil;
import com.twoandone.tms.realtime.utils.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;

/**
 * @packageName:com.twoandone.tms.realtime.app.func
 * @ClassName:DimBroadcastProcessFunction
 * @Description:
 * @author:
 * @date 2023/3/9 21:06
 **/
public class DimBroadcastProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private Map<String, TableProcess> map = new HashMap<>();


    @Override
    public void open(Configuration parameters) throws Exception {
//        System.out.println("open");
        Class.forName("com.mysql.cj.jdbc.Driver");
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/tms_dim_config?useSSl=false", "root", "000000");
        PreparedStatement prep = conn.prepareStatement("select * from tms_dim_config.table_process where  sink_type = 'dim'");
        ResultSet resultSet = prep.executeQuery();
        ResultSetMetaData metaData = resultSet.getMetaData();
        while (resultSet.next()) {
            JSONObject jsonObject = new JSONObject();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                String object = resultSet.getString(i);
                jsonObject.put(columnName, object);
            }
            TableProcess tableProcess = jsonObject.toJavaObject(TableProcess.class);
            String sinkTable = tableProcess.getSinkTable();
            map.put(sinkTable, tableProcess);
        }

    }

    public DimBroadcastProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void processElement(JSONObject jsonObject, ReadOnlyContext readOnlyContext, Collector<JSONObject> collector) throws Exception {
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
//        关联上的向下传
        String table = jsonObject.getString("table");
        TableProcess tableProcess = null;

//        System.out.println("====================");
        if ((tableProcess = broadcastState.get(table)) != null ||
                (tableProcess = map.get(table)) != null
        ) {

            JSONObject data = jsonObject.getJSONObject("data");
            String type = jsonObject.getString("type");

            filterColunm(data, tableProcess);

            data.put("sink_table", tableProcess.getSinkTable());
//            为后续redis做准备
            data.put("sink_type", type);
            System.out.println(data.toJSONString());

            collector.collect(data);
        }
    }

    private void filterColunm(JSONObject jsonObject, TableProcess tableProcess) {
        String sinkColumns = tableProcess.getSinkColumns();
        String[] colunms = sinkColumns.split(",");
        List<String> colunmList = Arrays.asList(colunms);
        Set<Map.Entry<String, Object>> entries = jsonObject.entrySet();
        entries.removeIf(entry -> !colunmList.contains(entry.getKey()));
    }

    /**
     * 广播流
     * 拿到数据之后，放到状态中
     *
     * @param jsonStr
     * @param context
     * @param collector
     * @throws Exception
     */
    @Override
    public void processBroadcastElement(String jsonStr, Context context, Collector<JSONObject> collector) throws Exception {
//"op":"c": {"before":null,"after":{"id":4,"username":"xzls","age":18},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1676889865000,"snapshot":"false","db":"gmall_config","sequence":null,"table":"t_user","server_id":1,"gtid":null,"file":"mysql-bin.000058","pos":1509,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1676889864853,"transaction":null}
//"op":"r": {"before":null,"after":{"id":1,"username":"zs","age":12},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"gmall_config","sequence":null,"table":"t_user","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1676889809851,"transaction":null}
//"op":"u": {"before":{"id":4,"username":"xzls","age":18},"after":{"id":4,"username":"jzls","age":18},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1676889901000,"snapshot":"false","db":"gmall_config","sequence":null,"table":"t_user","server_id":1,"gtid":null,"file":"mysql-bin.000058","pos":1796,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1676889900830,"transaction":null}
//"op":"d": {"before":{"id":4,"username":"jzls","age":18},"after":null,"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1676889930000,"snapshot":"false","db":"gmall_config","sequence":null,"table":"t_user","server_id":1,"gtid":null,"file":"mysql-bin.000058","pos":2099,"row":0,"thread":null,"query":null},"op":"d","ts_ms":1676889930755,"transaction":null}

        BroadcastState<String, TableProcess> broadcastState = context.getBroadcastState(mapStateDescriptor);

        JSONObject jsonObj = JSONObject.parseObject(jsonStr);

        String op = jsonObj.getString("op");
//        System.out.println("---------------------------");
        if ("d".equals(op)) {
            JSONObject data = jsonObj.getJSONObject("before");
            TableProcess tableProcess = JSON.toJavaObject(data, TableProcess.class);
            String sinkType = tableProcess.getSinkType();
            if ("dim".equals(sinkType)) {
                String sourceTable = tableProcess.getSourceTable();
                broadcastState.remove(sourceTable);
                map.remove(sourceTable);
            }

        } else {
            JSONObject data = jsonObj.getJSONObject("after");
            TableProcess tableProcess = data.toJavaObject(TableProcess.class);
            String sinkType = tableProcess.getSinkType();
//            System.out.println("dim1111111111111111");
            if ("dim".equals(sinkType)) {
                String sourceTable = tableProcess.getSourceTable();
                broadcastState.put(sourceTable, tableProcess);
//                要在这里创建表!
//                System.out.println("dim");
                checkTable(tableProcess);
            }
        }


    }

    private void checkTable(TableProcess tableProcess) {
//        在这里创建表，需要用到jdbc，那就直接写一个JDBC工具类
        String sinkTable = tableProcess.getSinkTable();
        String sinkColumns = tableProcess.getSinkColumns();
        String sinkPk = tableProcess.getSinkPk();
        if (StringUtils.isEmpty(sinkPk)) {
            sinkPk = "id";
        }
        String sinkExtend = tableProcess.getSinkExtend();

        StringBuilder sql = new StringBuilder(" create table  if not exists " + TmsConfig.PHOENIX_SCHEMA + "." + sinkTable + " ( ");
        String[] colunms = sinkColumns.split(",");
        for (int i = 0; i < colunms.length; i++) {
            if (sinkPk.equals(colunms[i])) {
                sql.append(colunms[i]).append(" varchar primary key,");
            } else {
                sql.append(colunms[i]).append(" varchar,");
            }
        }
        sql.deleteCharAt(sql.length() - 1).append(")");
        if (StringUtils.isNotEmpty(sinkExtend)) {
            sql.append(sinkExtend);
        }
        String sqlStr = sql.toString();
        System.out.println("创建表sql：" + sqlStr);
//        创建要用库名
        PhoenixUtil.excuteSql(sqlStr);
    }
}
